﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Diagnostics;
using System.Net;
using System.Net.Sockets;
using System.Text;
using System.Threading;
using System.Threading.Tasks;

namespace AMEE.TcpComm
{

    /// <summary>
    /// Hält Objekte zur Synchronisation von Reader und Sender Thread.
    /// </summary>
    internal struct RequestInfo
    {
        /// <summary>
        /// Queue für eingehende Requests
        /// </summary>
        public ConcurrentQueue<string> MessageQueue;

        /// <summary>
        /// The client
        /// </summary>
        public TcpClient Client;

    }

    /// <summary>
    /// Tcp Server. Arbeitet mit 2 Threads (read and respond) je Client Connection. Hält die Verbindung offen, bis Client abbricht oder Server beendet wird.
    /// </summary>
    public class TcpServer
    {
        #region Fields

        TcpListener _server;
        /// <summary>Flag, dass die Threads steuert.</summary>
        bool _doListening;
        System.Threading.Thread _listener;
        int _waitMillis;
        int _port;
        ManualResetEvent _waitRequest = new ManualResetEvent(false);

        ISet<TcpClient> _activeClients = new HashSet<TcpClient>();

        #endregion


        #region Properties

        Func<string, string> _processor;
        /// <summary>
        /// Gets or sets the Function, that processes the incoming Message and returns the request message.<para>
        /// Default is String.toUpper.</para>
        /// </summary>
        /// <value>
        /// The data processor function. (ignoring null)
        /// </value>
        public Func<string, string> DataProcessor {
            get { return _processor; }
            set {
                if (null == value)
                    return;
                _processor = value;
            }
        }


        /// <summary>
        /// Gets or sets the size of the buffer. Default is 4096 bytes.
        /// </summary>
        /// <value>
        /// The size of the buffer.
        /// </value>
        public int BufferSize { get; set; }

        /// <summary>
        /// Gets or sets the encoding. Default is ASCII
        /// </summary>
        /// <value>
        /// The encoding.
        /// </value>
        public Encoding Encoding { get; set; }

        /// <summary>
        /// Gets or sets a value indicating whether Server is running.
        /// </summary>
        /// <value>
        ///   <c>true</c> if server is running; otherwise, <c>false</c>.
        /// </value>
        public bool IsRunning { get; protected set; }

        /// <summary>
        /// Gets or sets a value indicating whether the connection automatically closes after response was sent<para>
        /// Default is False</para>
        /// </summary>
        /// <value>
        ///   <c>true</c> if close after respond; otherwise, <c>false</c>.
        /// </value>
        public bool CloseAfterResponse { get; set; }

        #endregion


        #region Constructor

        /// <summary>
        /// Initializes a new instance of the <see cref="TcpServer"/> class with default
        /// Encoding and DataProcessor.
        /// </summary>
        /// <param name="port">The port to accept Clients on.</param>
        public TcpServer(int port) {
            BufferSize = 4096;
            this.Encoding = Encoding.ASCII;
            DataProcessor = this.processData;
            IPAddress localAddr = IPAddress.Parse("127.0.0.1");
            _server = new TcpListener(localAddr, port);
            _port = port;
            _waitMillis = 5000; // 5 Sek. !!
        }

        #endregion


        #region Public

        /// <summary>
        /// Starts the server and handling of Tcp Requests.
        /// </summary>
        /// <exception cref="TcpCommException">Server konnte nicht gestartet werden.</exception>
        public void Start() {
            if (IsRunning) {
                Debug.WriteLine("Call of TcpEmbeddedServer.Start() while IsRunning.", "Info");
                return;
            }
            try {
                _server.Start();// Start listening for client requests.
                IsRunning = true;
                Debug.WriteLine("Server gestartet (Port " + _port + ")");
            }
            catch (SocketException e) {
                throw new TcpCommException("Server konnte nicht gestartet werden.", e);
            }
            _doListening = true;
            _listener = new Thread(() => handleTcpRequest());
            _listener.Name = "AcceptClientThread#" + _port + ":" + DateTime.Now.Millisecond;
            //listener.Priority = ThreadPriority.AboveNormal;
            _listener.Start();
        }

        /// <summary>
        /// Stops the server.
        /// </summary>
        /// <remarks>
        /// Stop have an async delay of 3 sec to handle a pending request.
        /// </remarks>
        public void Stop() {
            if (!IsRunning) {
                Debug.WriteLine("Call of TcpServer.Stop() while not IsRunning.", "Info");
                return;
            }
            // Stop listening for new clients.
            _doListening = false;
            Task.Run(() => {
                if (_listener != null) {
                    if (!_listener.Join(_waitMillis + 1000))
                        _listener.Interrupt(); //Abort();
                }
                _server.Stop();
                IsRunning = false;
            });
        }

        #endregion

        /// <summary>
        /// Process the data sent by the client.
        /// </summary>
        /// <param name="data">The data.</param>
        /// <returns>data.toUpper</returns>
        string processData(String data) {
            return data.ToUpper();
        }

        #region Thread Delegates

        /// <summary>
        /// Thread Delegat, der TCP Clients annimmt und an einen Serve-Thread übergibt.<para>
        /// Active Clients will be added to list here.</para>
        /// </summary>
        private void handleTcpRequest() {
            TcpClient client = null;
            while (_doListening && IsRunning) {
                Debug.WriteLine(Thread.CurrentThread.Name + " waiting for a connection...");
                try {
                    // Perform a blocking call to accept requests.
                    client = _server.AcceptTcpClient(); // throws SocketException, InvalidOperationException

                    var ip = (IPEndPoint)client.Client.RemoteEndPoint;
                    Debug.WriteLine(String.Format("New Client: {0}", ip.ToString()), "Info TcpEmbeddedServer");
                    if (_activeClients.Contains(client)) {
                        Debug.WriteLine("Client accepted, that is already in list of active clients!", "Warning TcpEmbeddedServer");
                        // und nu??
                    }
                    _activeClients.Add(client); // : bool
                    Thread handler = new Thread(cl => serveClient(cl));
                    handler.Name = "HandlerFor" + ip.ToString();
                    handler.Start(client);
                }
                catch (SocketException e) {
                    // SocketException - ggf. hat Client Verbindung unterbrochen
                    // oder dieser Thread wurde abgebrochen
                    Debug.WriteLine(Thread.CurrentThread.Name + "Exception (" + e.ErrorCode + "): " + e.Message, "Info TcpEmbeddedServer");
                }
                catch (Exception e) {
                    Debug.WriteLine(e.Message);
                }
            }
        }


        /// <summary>
        /// Thread Delegat. Liest solange _doListening=true vom Client-Stream und übergibt die Nachrichten an den ProcessAndRespond Thread.<para>
        /// Active Clients will be removed from list here.
        /// </para>
        /// </summary>
        /// <param name="clnt">The CLNT.</param>
        void serveClient(object clnt) {
            TcpClient client = clnt as TcpClient;
            NetworkStream stream;
            try {
                stream = client.GetStream(); // throws ObjectDisposedEx, InvalidOperationEx
                stream.ReadTimeout = _waitMillis; // magic number. Nach 5 Sek. abbrechen und prüfen, ob noch weitergelesen werden soll.
            }
            catch (Exception) {
                client.Close();
                _activeClients.Remove(client);
                Debug.WriteLine(Thread.CurrentThread.Name + "could not get stream.", "Info TcpEmbeddedServer");
                return;
            }

            var msgQueue = new ConcurrentQueue<string>(); // Queue für eingehende Requests

            // starte Thread für process data and send response
            var addr = (IPEndPoint)client.Client.RemoteEndPoint;
            var processor = new ResponseThread { Encoding = this.Encoding, DataProcessor = this.DataProcessor };
            Thread procThrd = new Thread(info => processor.ProcessAndRespond(info));
            procThrd.Name = "ProcessorFor" + addr.ToString();
            procThrd.Start(new RequestInfo { Client = client, MessageQueue = msgQueue });

            // Lese wiederholt die Requests des Clients:
            byte[] bytes = new byte[this.BufferSize];
            int readBytes;
            while (_doListening) {
                String data = "";
                try {
                    // read (with wait timeout) request
                    do {
                        //Debug.WriteLineIf(stream.CanRead, "lese message...", "Info TcpEmbeddedServer");
                        //Debug.WriteLineIf(!stream.CanRead, "Kann nicht lesen!", "Info TcpEmbeddedServer");
                        readBytes = stream.Read(bytes, 0, bytes.Length);
                        data += this.Encoding.GetString(bytes, 0, readBytes);
                    } while (stream.DataAvailable && readBytes != 0);
                    Debug.WriteLineIf(data != String.Empty, String.Format("Srv Received: \"{0}\" ({1})", data, Thread.CurrentThread.Name), "Info TcpEmbeddedServer");
                    if (0 == readBytes) {
                        // Client hat Kommunikatin beendet!?!
                        Debug.WriteLineIf(readBytes == 0, String.Format("Client hat beendet. (read byte=0) ({0})", Thread.CurrentThread.Name), "Info TcpEmbeddedServer");
                        break;
                    }
                    // pass on data for processing and response)
                    if (data != String.Empty)
                        msgQueue.Enqueue(data);
                }
                catch (System.IO.IOException ioEx) {
                    // Client weg? ...
                    if (ioEx.InnerException.HResult == -2146232800) {
                        Debug.WriteLine(String.Format("Verbindung zum Cient vermutlich abgebrochen ({0})", Thread.CurrentThread.Name), "Info TcpEmbeddedServer");
                        break; // stop loop
                    }
                    if (ioEx.InnerException.HResult == -2147467259) {
                        Debug.WriteLine(String.Format("Timeout beim lesen ({0})", Thread.CurrentThread.Name), "Info TcpEmbeddedServer");
                        continue; // stop loop
                    }
                }
                catch (Exception e) {
                    Debug.WriteLine(e.Message);
                }
            }
            // clean up:
            // stop correspondig ResponseThread
            processor.RequestStop();
            if (!procThrd.Join(5000)) {
                procThrd.Abort();
            }
            // close my client (of this Thread):
            client.Close();
            _activeClients.Remove(client);
        }



        /// <summary>
        /// Thread Klasse (ProcessAndRespond(RequestInfo)).<para>
        /// Abbruch des Threads mit RequestStop().
        /// </para>
        /// </summary>
        /// <remarks>
        /// Nutzung:
        /// <c>
        /// var processor = new ResponseThread { Encoding= this.Encoding, DataProcessor = this.DataProcessor};
        /// Thread procThrd = new Thread(info => processor.ProcessAndRespond(info));
        /// procThrd.Start(new RequestInfo { Client = client, MessageQueue = msgQueue });
        /// ...
        /// processor.RequestStop();
        /// procThrd.Join()
        /// </c></remarks>
        private class ResponseThread
        {

            private volatile bool _shouldStop;
            int _waitMillis;

            /// <summary>
            /// Gets or sets the encoding. Default is ASCII
            /// </summary>
            /// <value>
            /// The encoding.
            /// </value>
            public Encoding Encoding { get; set; }

            /// <summary>
            /// Gets or sets the Function, that processes the incoming Message and returns the request message.<para>
            /// Default is String.toUpper.</para>
            /// </summary>
            /// <value>
            /// The data processor function. (ignoring null)
            /// </value>
            public Func<string, string> DataProcessor { get; set; }


            public ResponseThread() {
                _waitMillis = 5000;
            }

            public void RequestStop() {
                _shouldStop = true;
            }


            /// <summary>
            /// Thread Delegat. Solange _doListening=true, liese von MsgQueue, verarbeite Nachricht und sende Antwort.
            /// </summary>
            /// <param name="requestInfo">The request information.</param>
            public void ProcessAndRespond(object requestInfo) {
                RequestInfo info = (RequestInfo)requestInfo;
                TcpClient client = info.Client;
                var msgQueue = info.MessageQueue;
                String msg;
                String response;
                NetworkStream stream;
                try {
                    stream = client.GetStream(); // throws ObjectDisposedEx, InvalidOperationEx
                    stream.WriteTimeout = _waitMillis;  // magic number. Nach 5 Sek. abbrechen und prüfen, ob noch weitergelesen werden soll.
                }
                catch (Exception) {
                    //client.Close();
                    //_activeClients.Remove(client);
                    Debug.WriteLine(Thread.CurrentThread.Name + "could not get stream.", "Info TcpEmbeddedServer");
                    return;
                }

                while (!_shouldStop) {
                    // wait for a message
                    if (!msgQueue.TryDequeue(out msg)) {
                        Thread.Sleep(1000);
                        continue;
                    }
                    try {
                        // Process Request
                        response = this.DataProcessor(msg);
                        // Respond:
                        byte[] resp = this.Encoding.GetBytes(response);
                        Debug.WriteLineIf(!stream.CanWrite, Thread.CurrentThread.Name + ". Auf Stream kann nicht geschrieben werden!", "Warning TcpEmbeddedServer");
                        stream.Write(resp, 0, resp.Length);
                        Debug.WriteLine(String.Format("Srv sent: \"{0}\" ({1})", response, Thread.CurrentThread.Name), "Info TcpEmbeddedServer");
                    }
                    catch (EncoderFallbackException efEx) {
                        Debug.WriteLine(efEx.Message, "Warning TcpEmbeddedServer");
                    }
                    catch (Exception e) {
                        Debug.WriteLine(e.Message);
                        // continue;
                    }
                }
            }
        }

        #endregion


    }
}
